package com.gitee.ixtf.rsocket.agent

import cn.hutool.core.util.RandomUtil
import cn.hutool.log.Log
import com.gitee.ixtf.core.J
import com.gitee.ixtf.rsocket.BROKER_HOST
import com.gitee.ixtf.rsocket.BROKER_PORT
import com.gitee.ixtf.rsocket.IRSocketAgent
import com.gitee.ixtf.rsocket.RSocketStatus
import com.gitee.ixtf.rsocket.broker.SetupRequest
import io.cloudevents.core.builder.CloudEventBuilder
import io.rsocket.*
import io.rsocket.core.RSocketClient
import io.rsocket.core.RSocketConnector
import io.rsocket.frame.decoder.PayloadDecoder
import io.rsocket.transport.ClientTransport
import io.rsocket.transport.netty.client.TcpClientTransport
import io.vertx.ext.auth.authentication.Credentials
import java.net.URI
import java.time.Duration
import kotlin.properties.Delegates
import kotlin.time.Duration.Companion.milliseconds
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.*
import kotlinx.coroutines.reactor.mono
import reactor.core.publisher.Mono
import reactor.core.scheduler.Schedulers
import reactor.util.retry.Retry

private val EMPTY_RSOCKET: RSocket = object : RSocket {}

abstract class RSocketAgent : SocketAcceptor, IRSocketAgent {
  protected val log: Log by lazy { Log.get(this::class.java) }
  protected open var serviceRSocket: RSocket = EMPTY_RSOCKET
  @field:Volatile private var client: RSocketClient? = null
  private var status by
      Delegates.observable(RSocketStatus.INIT) { _, oldV, newV ->
        log.warn("$oldV -> $newV")
        if (newV == RSocketStatus.DISCONNECT) tryConnect()
        if (newV == RSocketStatus.DOWN) client?.dispose()
        onStatus(oldV, newV)
      }

  override suspend fun cloudEventBuilder(): CloudEventBuilder =
      CloudEventBuilder.v1().withId(J.objectId()).withSource(URI("client"))

  override suspend fun rSocketClient() =
      flow {
            while (true) {
              emit(
                  when (status) {
                    RSocketStatus.UP -> client
                    RSocketStatus.INIT,
                    RSocketStatus.DISCONNECT -> tryConnect().let { null }
                    else -> null
                  })
              delay(RandomUtil.randomLong(100, 500).milliseconds)
            }
          }
          .filterNotNull()
          .first()

  override fun accept(setup: ConnectionSetupPayload, sendingSocket: RSocket): Mono<RSocket> {
    sendingSocket
        .onClose()
        .subscribeOn(Schedulers.boundedElastic())
        .doAfterTerminate { if (status == RSocketStatus.UP) status = RSocketStatus.DISCONNECT }
        .subscribe()
    return Mono.just(serviceRSocket).doOnSuccess { status = RSocketStatus.UP }
  }

  @Synchronized
  private fun tryConnect() {
    if (RSocketStatus.DOWN == status ||
        RSocketStatus.IDLE == status ||
        RSocketStatus.CONNECTING == status)
        return
    if (client?.isDisposed == false) client?.dispose()
    client = buildRSocketClient().apply { connect() }
    status = RSocketStatus.CONNECTING
  }

  protected open fun buildRSocketClient(): RSocketClient =
      RSocketClient.from(
          RSocketConnector.create()
              .payloadDecoder(PayloadDecoder.ZERO_COPY)
              .acceptor(this)
              .setupPayload(buildSetupRequest())
              .reconnect(buildRetry())
              .connect(buildClientTransport()))

  protected open fun buildSetupRequest(): Mono<Payload> = mono {
    SetupRequest.client(buildCredentials())
  }

  protected abstract suspend fun buildCredentials(): Credentials

  protected open fun buildClientTransport(): ClientTransport =
      TcpClientTransport.create(BROKER_HOST, BROKER_PORT)

  protected open fun onStatus(oldV: RSocketStatus, newV: RSocketStatus) = Unit

  protected open fun buildRetry(): Retry =
      Retry.backoff(Long.MAX_VALUE, Duration.ofSeconds(1))
          .scheduler(Schedulers.boundedElastic())
          .minBackoff(Duration.ofSeconds(1))
          .maxBackoff(Duration.ofSeconds(3))

  override fun dispose() {
    status = RSocketStatus.DOWN
  }

  override fun isDisposed(): Boolean = client?.isDisposed ?: (status == RSocketStatus.DOWN)

  override fun onClose(): Mono<Void> = client?.onClose() ?: serviceRSocket.onClose()
}
